package com.finger.gulimall.ware.config;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import com.finger.common.constant.MqStockConstant;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @ClassName MyRabbitConfig
 * @Description
 * @Author finger
 * @Date 2022/7/3 11:23
 * @Version 1.0
 */

@Configuration
public class RabbitConfig {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 配置一个Rabbit发送消息自动转换Json格式
     * @return Jackson2JsonMessageConverter
     */
    @Bean
    public MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 定制化RabbitTemplate
     * <p>
     * 生产端可靠消息投递
     * 1.服务器收到消息就回调
     *      1.配置文件配置spring.rabbitmq.publisher-confirms=true
     *      2.rabbitTemplate设置ConfirmCallback回调函数
     * 2.交换机将消息放入队列失败后回调
     *      1.配置文件配置spring.rabbitmq.publisher-returns
     *      2.rabbitTemplate设置ReturnCallback回调函数
     * 3.消费者确认（保证每个消息被正确消费，此时才可以broker删除这个消息）。
     *      spring.rabbitmq.listener.simple.acknowledge-mode 手动签收消息
     *      1.默认是自动确认的，只要消息接收到，客户端会自动确认，服务端就会移除这个消息
     *          问题：
     *              我们收到很多消息，自动回复给服务器ack，只有一个消息处理成功，宕机了。发生消息丢失问题
     *              消费者手动确认模式。只要我们没有手动告诉服务器ack，货物被签收。没有ack，消息就一直是unacked状态。即使消费者宕机，消息不会丢失，会重新变成Ready，下次还会被消费端消费
     *      2，如何签收
     *          channel.basicAck(deliveryTag,false); 接收消息手动应答
     *          channel.basicNack(deliveryTag,false,true); 拒收消息
     */
    @PostConstruct
    public void initRabbitTemplate() {

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             *  @param correlationData correlationData 当前消息的唯一关联数据
             *       （唯一表示id，可以在发送消息的时候设置，然后存入本地数据库，消息放入服务器之后回掉函数中更新数据库中该消息是否发送成功）
             *  @param success 是否发送成功
             *  @param reason 失败原因，没出错就null
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean success, String reason) {
                System.out.println("消息进入服务：correlationData:" + correlationData + "；success：" + success + "：原因：" + reason);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * 只要消息没有投递到指定的队列，就触发这个失败回调
             * @param message 投递失败的消息详细信息
             * @param replyCode 回复的状态码
             * @param replyText 回复的文本内容
             * @param exchange 当时这个消息发给哪个交换机
             * @param routingKey 当时这个消息用的哪个路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

                System.out.println("消息进入队列失败：message:" +
                        message + "；replyCode：" + replyCode + "：原因：" + replyText+"；exchange："+exchange+"；routingKey："+routingKey);
            }
        });
    }

    @Bean
    public Exchange orderEventExchange(){
        return new TopicExchange(MqStockConstant.STOCK_EVENT_EXCHANGE, true, false);
    }

    @Bean
    public Queue orderDelayDueue(){
        Map<String,Object> arguments = new HashMap<>();
        /**
         * x-dead-letter-exchange: user.order.exchange
         * x-dead-letter-routing-key: order
         * x-message-ttl: 60000
         */
        arguments.put("x-dead-letter-exchange",MqStockConstant.STOCK_EVENT_EXCHANGE);
        arguments.put("x-dead-letter-routing-key",MqStockConstant.STOCK_RELEASE_STOCK_QUEUE);
        arguments.put("x-message-ttl",120000);
        return new Queue(MqStockConstant.STOCK_DELAY_QUEUE, true, false, false,arguments);

    }

    @Bean
    public Queue orderReleaseOrderQueue(){

        return new Queue(MqStockConstant.STOCK_RELEASE_STOCK_QUEUE, true, false, false,null);

    }

    @Bean
    public Binding orderCreateOrder(){
        return new Binding(MqStockConstant.STOCK_DELAY_QUEUE, Binding.DestinationType.QUEUE,MqStockConstant.STOCK_EVENT_EXCHANGE,MqStockConstant.STOCK_DELAY_QUEUE,null);
    }

    @Bean
    public Binding orderReleaseOrder(){
        return new Binding(MqStockConstant.STOCK_RELEASE_STOCK_QUEUE, Binding.DestinationType.QUEUE,MqStockConstant.STOCK_EVENT_EXCHANGE,MqStockConstant.STOCK_RELEASE_STOCK_QUEUE,null);
    }
}
